package org.dreamwork.cai.io.impl;

import com.google.gson.Gson;
import org.dreamwork.cai.core.BroadcastPayload;
import org.dreamwork.cai.io.AbstractRouteBroadcaster;
import org.dreamwork.cai.proxy.IServiceResolver;
import org.dreamwork.cai.tools.Misc;
import org.dreamwork.concurrent.Looper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.util.List;

public class UdpRouteBroadcaster extends AbstractRouteBroadcaster {
    private final Logger logger = LoggerFactory.getLogger (UdpRouteBroadcaster.class);

    private static final String LOOP_NAME = "UdpRouteBroadcaster";
    private static final int    BUFF_SIZE = 8192;   // 8k

    private MulticastSocket server;
    private MulticastSocket broadcaster;
    private InetAddress target;
    private boolean running = true;

    public UdpRouteBroadcaster (IServiceResolver resolver, int senderPort) {
        super (resolver, senderPort);
        if (logger.isTraceEnabled ()) {
            logger.trace ("+++++++++++++++++++++++++++++++++++++++");
            logger.trace ("+ sender port = {}", senderPort);
            logger.trace ("+++++++++++++++++++++++++++++++++++++++");
        }
    }

    @Override
    protected void bind0 (int port) throws IOException {
        // 启动 udp 组播
        this.port = port;
        SocketAddress local = new InetSocketAddress ("0.0.0.0", port);
        server = new MulticastSocket (local);
        server.setReuseAddress (true);
        logger.info ("Udp Route Broadcaster listen on {}", server);

        broadcaster = new MulticastSocket (senderPort);
        target      = InetAddress.getByName ("225.0.0.1");

        List<NetworkInterface> list = Misc.findAllValidNetworkInterfaces ();
        SocketAddress remote = new InetSocketAddress (target, port);
        for (NetworkInterface ni : list) {
            server.joinGroup (remote, ni);
            broadcaster.joinGroup (remote, ni);
        }

        Looper.invokeLater (() -> {
            logger.info ("UDP Router Broadcaster listened on {}", port);
            Looper.create (LOOP_NAME, 16);
            while (running) {
                byte[] buff = new byte[BUFF_SIZE];       // 8k buff
                DatagramPacket packet = new DatagramPacket (buff, BUFF_SIZE);
                try {
                    server.receive (packet);
                    Looper.runInLoop (LOOP_NAME, new UdpRouteReceiver (packet,senderPort, resolver));
                } catch (IOException ex) {
                    logger.warn (ex.getMessage (), ex);
                }
            }
            logger.info ("UDP Route Broadcaster shutdown.");
        });
    }

    @Override
    protected void unbind0 () {
        running = false;
        logger.info ("udp Route Broadcaster shutting down...");
    }

    @Override
    protected void send (BroadcastPayload payload) throws IOException {
        Gson g = new Gson ();
        String content = g.toJson (payload);
        byte[] buff = content.getBytes (StandardCharsets.UTF_8);
        DatagramPacket packet = new DatagramPacket (buff, buff.length, target, port);
        broadcaster.send (packet);
    }
}